/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl.disk;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * File for write-ahead logging.
 * @author Alex Antonau
 * */
public class LogFile {
	
	public interface AtomicWrite {
		public void write(DataOutputStream dos) throws IOException;
	}
	
	private final int writeBufferSize;
	private File file;
	private DataOutputStream outputStream;
	private long length;
	private final ReentrantReadWriteLock lock;
	
	public LogFile(File file, int writeBufferSize) throws IOException {
		super();
		this.file = file;
		this.length = file.length();
		this.writeBufferSize = writeBufferSize;
		this.lock = new ReentrantReadWriteLock();
		if(!file.exists()){
			File backup = getBackup();
			if(backup.exists()) backup.renameTo(file);
		}
		//open file to insure that there is only one writer open.
		getOutputStream();
	}
	
	public LogFile(File file) throws IOException {
		this(file, 512);
	}

	public ReentrantReadWriteLock getLock() {
		return lock;
	}
	
	private DataOutputStream getOutputStream() throws IOException {
		if(outputStream == null){
			FileOutputStream fos = new FileOutputStream(file, true);
			BufferedOutputStream bos = new BufferedOutputStream(fos, writeBufferSize);
			outputStream = new DataOutputStream(bos);
			this.length = file.length();
		}
		return outputStream;
	}
	/**
	 * Returns the file object used to create this log file.
	 * @return a <code>File</code> which represents the log file.
	 */
	public File getFile() {
		lock.readLock().lock();
		try{
			return file;
		} finally{
			lock.readLock().unlock();
		}
	}
	private File getBackup(){
		String path = file.getPath();
		File backup=new File(path+".bak");
		return backup;
	}
	
	public void substituteFile(File newFile) throws IOException {
		lock.writeLock().lock();
		try{
			if(outputStream!=null) try{ outputStream.close(); } catch (Exception e) {}
			outputStream = null;
			String path = file.getPath();
			File backup=getBackup();
			backup.delete();
			if(!file.renameTo(backup)) throw new IOException("Unable to rename "+file+" to "+backup);
			file = new File(path);
			if(!newFile.renameTo(file))throw new IOException("Unable to rename "+newFile+" to "+file);
			backup.delete();
			this.length = file.length();
		} finally{
			lock.writeLock().unlock();
		}
	}

	/**
	 * Flushes write buffers to the disk.
	 * The method is used when preparing to commit, to ensure all the updates
	 * have been written to the log.
	 * 
	 * @exception IOException
	 *                if an I/O error occurs
	 */
	public void flush() throws IOException {
		if(outputStream != null) {
			lock.writeLock().lock();
			try{
				outputStream.flush();
			} finally {
				lock.writeLock().unlock();
			}
		}
	}
	
	public void close() throws IOException {
		if(outputStream!=null){
			lock.writeLock().lock();
			try{
				try{ outputStream.close(); } catch (Exception e) {}
				outputStream = null;
			} finally{
				lock.writeLock().unlock();
			}
		}
	}
		
	public void delete() throws IOException {
		lock.writeLock().lock();
		try{
			if(outputStream!=null) try{ outputStream.close(); } catch (Exception e) {}
			outputStream = null;
			file.delete();
			getBackup().delete();
		} finally{
			lock.writeLock().unlock();
		}
	}
	
	public long getLength() throws IOException {
		lock.readLock().lock();
		try{
			return length;
		} finally{
			lock.readLock().unlock();
		}
	}
	/**
	 * Appends data block to the log file.
	 * @param data data to write to the log file
	 * @throws IOException if an I/O error occurs
	 * @return offset of the data block in the log file
	 */
	public long write(byte[] data, int off, int len) throws IOException {
		lock.writeLock().lock();
		boolean success = false;
		try{
			DataOutputStream dos = getOutputStream();
			long dataPosition = length;
			dos.write(data, off, len);
			length += len;
			success = true;
			return dataPosition;
		} finally{
			//if a write failed then restart & recovery is needed
			if(!success) try{ outputStream.close(); }catch (Exception e) {	}
			lock.writeLock().unlock();
		}
	}
	
	public void write(AtomicWrite op) throws IOException{
		lock.writeLock().lock();
		boolean success = false;
		try{
			DataOutputStream dos = getOutputStream();
			op.write(dos);
			dos.flush();
			length = file.length();
			success = true;
		} finally{
			//if a write failed then restart & recovery is needed
			if(!success) try{ outputStream.close(); }catch (Exception e) {	}
			lock.writeLock().unlock();
		}
	}

}
